package timeandwindow;

import com.atguigu.pojo.Event;
import com.atguigu.pojo.WordCount;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import util.SourceUtil;

import java.time.Duration;
import java.util.HashMap;

/**
 * 案例需求：统计10秒内每用户点击次数, 输出包含窗口信息
 * //窗口：按键（按用户分区）分区事件时间滚动窗口
 * 全窗口函数：
 *    ProcessWindowFunction:将进入到窗口的数据收集起来，等到窗口触发计算时，
 *    将所收集的数据一次性进行计算，输出结果。支持获取窗口信息
 *    四个泛型：
 *       <IN> The type of the input value. 输入类型
 *       <OUT> The type of the output value. 输出类型
 *       <Key> The type of the key. Key的类型
 *       <W> The type of Window that this window function can be applied on. 窗口类型
 *            时间窗口：TimeWindow
 *            计数窗口：GlobalWindow  
 *   
 */
public class Flink09_ProcessWindowFunction {
     public static void main(String[] args) {
             StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
             env.setParallelism(1);
             
             //对接数据源，设置水位线
         SingleOutputStreamOperator<Event> ds = env
                 .fromSource(SourceUtil.getSource(), WatermarkStrategy.noWatermarks(), "dataGenSource")
                 .assignTimestampsAndWatermarks(
                         WatermarkStrategy.<Event>forBoundedOutOfOrderness(Duration.ZERO)
                                 .withTimestampAssigner(
                                         (event, ts) -> event.getTs()
                                 )
                 );
         
         ds.print("INPUT");
         
         //统计10s内每用户点击次数，输出包含窗口信息
         //窗口类型： 按键分区事件时间滚动窗口
         
         ds
                 .map( event -> new WordCount(event.getUser(),1L))
                 .keyBy(WordCount::getWord)
                 .window(
                         TumblingEventTimeWindows.of(Time.seconds(10))
                 )
                 .process(

                         /**
                          * ProcessWindowFunction<IN, OUT, KEY, W extends Window>
                          * WordCount:输入类型
                          * String：输出类型
                          * String：key
                          * TimeWindow：窗口类型
                          */
                         new ProcessWindowFunction<WordCount, String, String, TimeWindow>() {
                             /**
                              *
                              * @param key :key类型
                              * @param context ：上下文
                              * @param elements :窗口中收集到的数据
                              * @param out
                              * @throws Exception
                              */
                             @Override
                             public void process(String key, ProcessWindowFunction<WordCount, String, String, TimeWindow>.Context context, Iterable<WordCount> elements, Collector<String> out) throws Exception {

                                 HashMap<String, Long> userViewCount = new HashMap<>();
                                 for (WordCount element : elements) {
                                     //elements:元素
                                     Long count = userViewCount.getOrDefault(element.getWord(), 0L);//初始化
                                     //存到map中
                                     userViewCount.put(element.getWord(),count+element.getCount());
                                 }
                                 //获取窗口信息
                                 long windowStart = context.window().getStart();
                                 long windowEnd = context.window().getEnd();

                                 //输出
                                 userViewCount.forEach((user,count) -> out.collect("窗口 ：["+ windowStart + ","+ windowEnd + "),user:" +user+",count:" + count));
                             }

                         }
                 )
                 .print("WINDOW");

         try {
                 env.execute();
             } catch (Exception e) {
                 throw new RuntimeException(e);
             }
         }
}
